package com.gitee.broadcast

import java.util

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
/* TODO 广播变量,分布式文件缓存,累加器的总结
1.广播变量:是将变量(小变量)发给各个TaskManager,然后各个TaskManager中的subTask可以去各自的TaskManager中获取广播变量

2.分布式文件缓存:是将文件(小文件)发给各个TaskManager,然后各个TaskManager中的subTask可以去各自的TaskManager中获取文件

3.累加器:在全局什么一个累加器变量,然后各个SubTask可以将数值累加到该累加器中
 */


/*
  1. 确定要广播的对象
  2. 调用RichMapFunction函数里的getRuntimeContext方法获取广播变量
  3. 将对象广播出去withBroadcastSet(对象,别名)
  都必须扩展RichFunction类
 */
object BroadCastDemo {
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    //准备数据
    val studentDataSet: DataSet[(Int, String)] = env.fromCollection(List((1, "张三"), (2, "李四"), (3, "王五")))
    val scoreDataSet: DataSet[(Int, String, Int)] = env.fromCollection(List((1, "语文", 50), (2, "数学", 70), (3, "英文", 86)))

    //处理数据
    //明确将谁进行广播 -- 这里我们肯定是将不太常变化studentDataSet广播
    //那么就意味着scoreDataSet中要使用广播变量, TODO 而使用广播需要使用 getRuntimeContext 方法
    // 而该方法在RichMapFunction中,所以map方法中应该传入new RichMapFunction
    val studentBroadcast: DataSet[(String, String, Int)] = scoreDataSet.map(new RichMapFunction[(Int, String, Int), (String, String, Int)] {
      //准备map集合用来接收获取到的广播变量的值
      //Map[学生id,学生姓名]
      var studentMap: Map[Int, String] = _

      //open方法只会在各个SubTask开始的时候执行一次,所以在这里比较适合获取广播
      override def open(parameters: Configuration): Unit = {
        // 这里我们需要手动给指定一下类型
        val list: util.List[(Int, String)] = getRuntimeContext.getBroadcastVariable("studentBroadcast")
        //TODO 将Java的List转为Scala的Map, 为什么要转换呢?
        import scala.collection.JavaConverters._
        studentMap = list.asScala.toMap
      }

      override def map(value: (Int, String, Int)): (String, String, Int) = {
        val studentId: Int = value._1
        val studentName = studentMap(studentId)
        (studentName, value._2, value._3)
      }

      //将DataSet广播出去
    }).withBroadcastSet(studentDataSet, "studentBroadcast")
    studentBroadcast.print()
  }

}
